package com.atguigu.flink.datastramapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by Smexy on 2023/2/27
 *      自定义sink，以写出到Mysql
 *
 *
 *      env.addSink(SinkFunction x)
 *
 */
public class Demo5_CustomSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(new MySink());
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    /*
            SinkFunction: 只有写出功能。
            RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction: 既有写出功能，还有生命周期方法。

     */
    private static class MySink extends RichSinkFunction<WaterSensor>
    {

        private Connection connection;
        //保存每个传感器最新的数据信息
        private String sql = "replace into sensor values(?,?,?)";
        private PreparedStatement ps;

        //写出，每条数据调用一次
        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

            ps.setString(1,value.getId());
            ps.setLong(2,value.getTs());
            ps.setInt(3,value.getVc());
            
            ps.execute();

        }

        //只在task创建时，运行一次
        @Override
        public void open(Configuration parameters) throws Exception {


            connection = DriverManager.getConnection(
                "jdbc:mysql://hadoop104:3306/220926?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true",
                "root",
                "000000"
            );

            ps = connection.prepareStatement(sql);
        }

        //只在task停止时，运行一次
        @Override
        public void close() throws Exception {

            if (connection != null){
                connection.close();
            }
        }
    }
}
